package dws

import java.io.InputStream
import java.net.URL
import java.util

import org.apache.spark.sql.{DataFrame, Row, SaveMode, SparkSession}
import org.apache.spark.{SparkConf, SparkContext}

import scala.io.{BufferedSource, Source}

object dws_order {

  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession
      .builder()
      .appName("dws.dws_order")
      .config("spark.sql.parquet.writeLegacyFormat", true)
      .config("spark.sql.hive.convertMetastoreParquet","false")
      .getOrCreate()

    import spark.implicits._

    //1. 清除dws_order表数据
    spark.sql("truncate table tmp.dws_order")

    //2. 配置执行参数
    spark.sql("set hive.exec.dynamici.partition=true")
    spark.sql("set hive.exec.dynamic.partition.mode=nonstrict")
//    spark.sql("set spark.sql.adaptive.enabled=true")
    spark.sql("set spark.sql.shuffle.partitions=100")
    //3. 插入全量dws数据
    val file_path ="hdfs://dpcdh100:8020/user/hive/warehouse/tmp.db/dws_order"
    val dws_order_df: DataFrame = spark.sql(
      """
        --INSERT into table tmp.dws_order partition(dt='0000-00-00')
        select
              '' order_detail_key
              ,order_info.order_id order_id-- 订单i
              ,order_info.order_code order_code-- 订单
              ,order_info.SOURCE SOURCE-- 订单类型 是否需要转义成汉
              ,order_info.open_phone
              ,order_info.student_name
              ,order_info.leader_num
              ,order_info.create_time
              ,order_info.leader_category_code
              ,order_info.setmeal_name_category_code
              ,order_info.leader_present_num
              ,order_info.leader_present_name
              ,order_info.user_id
              ,order_info.dp_account
              ,order_pay.first_pay_date
              ,order_pay.first_day_money
              ,order_pay.first_month_money
              ,order_pay.first_pay_time
              ,lead(first_pay_time,1,'2099-01-01 00:00:00') over(distribute by user_id,order_info.leader_category_code sort by first_pay_time)  order_end_time
              ,order_pay.first_pay_money
              ,CASE
                  WHEN order_pay.first_day_money > order_info.receivable_price_handle THEN '超额订单'
                  WHEN order_pay.first_day_money = order_info.receivable_price_handle THEN '全款订单'
                  WHEN order_pay.first_day_money >= order_pay.standard_price  AND order_pay.first_day_money < order_info.receivable_price_handle THEN '开课订单'
                  WHEN order_pay.first_day_money < order_pay.standard_price THEN '定金订单'
                  else '未完成支付订单'
              END as  order_type1
              -- #########
              ,CASE WHEN order_pay.reach_standard_time < '2099-12-31 23:59:59' THEN 'Y' ELSE'N'  END is_positive
              ,order_pay.reach_standard_time as positive_time
              ,order_info.user_group
              ,order_info.drop_out
              ,order_info.invoice_money_handle   -- 开票金额目前没有看到数据
              ,order_info.invoice_prestore_money_handle
              ,order_info.favourable_use_price_handle
              ,order_info.favourable_price_handle
              ,order_info.result_price_handle
              ,order_pay.all_pay
              ,abs(order_pay.return_pay)  as return_pay
              ,order_pay.last_pay_time
              ,order_info.chose_add_price_handle
              ,order_info.order_type
              ,order_info.flow_account
              ,order_info.flow_account_type
              ,order_info.payment_status
              ,order_info.open_course_status
              ,order_info.order_attribute
              ,order_info.dy_account
              ,order_info.dy_num
              ,order_info.age_group
              ,order_info.sex
              ,order_info.receivable_price_handle
              ,order_info.setmeal_id
              ,order_info.setmeal_name
              ,order_info.regular_status
              ,order_info.regular_time
              ,order_info.student_city
              ,order_info.student_provinces
              ,case when order_pay.reach_refund_standard_time is null then '2099-12-31 23:59:59' else order_pay.reach_refund_standard_time end  as join_return_reach_standard_time
              ,case when order_pay.reach_refund_standard_time is null or order_pay.reach_refund_standard_time = '2099-12-31 23:59:59'  then 'N' else 'Y' end as  join_return_reach_standard_status
              ,order_info.identity_card_num
              ,order_info.father_code
              ,order_info.before_code
              ,order_info.is_main
              ,order_info.father_id
              ,order_info.promotions_type
              ,order_info.is_foreign
              ,'Y' FLAG
              ,from_unixtime(unix_timestamp(),'yyyy-MM-dd HH:mm:ss') AS op_time
              ,'$exe_id' execid
              ,substring(order_pay.first_pay_date,1,10) as dt
          from
          (
              SELECT
                  ID order_id-- 订单i
                  ,code order_code-- 订单
                  ,SOURCE SOURCE-- 订单类型 是否需要转义成汉
                  ,open_phone
                  ,student_name
                  ,leader_num
                  ,create_time
                  ,setmeal_name_category_code
                  ,student_city
                  ,student_provinces
                  ,leader_category_code
                  ,user_id
                  ,dp_account
                  ,order_type
                  ,order_attribute
                  ,dy_account
                  ,dy_num
                  ,age_group
                  ,setmeal_id
                  ,setmeal_name
                  ,regular_status
                  ,regular_time
                  ,receivable_price_handle
                  ,open_course_status
                  ,flow_account
                  ,flow_account_type
                  ,chose_add_price_handle
                  ,result_price_handle
                  ,favourable_use_price_handle
                  ,favourable_price_handle
                  ,invoice_money_handle-- 开票金额目前没有看到数据
                  ,invoice_prestore_money_handle
                  ,drop_out
                  ,user_group
                  ,payment_status
                  ,leader_present_num
                  ,leader_present_name
                  ,sex
                  ,identity_card_num
                  ,father_code
                  ,before_code
                  ,is_main
                  ,father_id
                  ,promotions_type
                  ,is_foreign
              FROM
              dwd.dwd_order_info
              WHERE
                  FLAG = 'Y'
                  AND
                  (msg_status = 'Y' or msg_status ='y')
          ) order_info
          left join
          (
              select
                  order_code
                  ,leader_category_code
                  ,return_pay
                  ,first_pay_date
                  ,first_day_money
                  ,first_month_money
                  ,first_pay_time
                  ,first_pay_money
                  ,last_pay_time
                  ,sum_pay_money as all_pay
                  ,min(reach_standard_time) reach_standard_time
                  ,min(reach_refund_standard_time) reach_refund_standard_time
                  ,standard_price

              from
              (
                  select
                      a.order_code
                      ,leader_category_code
                      ,pay_create_time
                      ,return_pay
                      ,first_pay_date
                      ,first_day_money
                      ,first_month_money
                      ,first_pay_time
                      ,first_pay_money
                      ,last_pay_time
                      ,sum_pay_money
                      ,CASE WHEN a.pay_add >= regular.standard_price or a.pay_add >= a.receivable_price  THEN a.pay_create_time ELSE '2099-12-31 23:59:59' END reach_standard_time
                      ,CASE WHEN (a.pay_refund_add >= regular.standard_price or a.pay_refund_add >= a.receivable_price) and a.pay_create_time is not null THEN a.pay_create_time ELSE '2099-12-31 23:59:59' END reach_refund_standard_time
                      ,standard_price
                  from
                      (
                          select
                              order_code
                              ,pay_create_time
                              ,dd.receivable_price
                              ,dd.setmeal_name_category_code leader_category_code
                              ,sum(case when type = 'refund' then payment_price else 0 end) over(partition by order_code) return_pay
                              ,sum(payment_price) over(partition by order_code order by pay_create_time)  pay_refund_add
                              ,sum(case when type = 'pay' then payment_price else 0 end) over(partition by order_code order by pay_create_time) pay_add

                          from
                          (
                              select
                                  distinct order_code
                                  ,payment_price/100 payment_price
                                  ,pay_create_time
                                  ,leader_category_code
                                  ,'pay' as type
                              from
                                  ods.ods_of_payment pay_add
                              where
                                  (pay_add.msg_status = 'Y' or pay_add.msg_status = 'y')
                                  and pay_add.order_code is not null
                                  and pay_add.order_code != ''
                              union all
                              select
                                  distinct order_code
                                  ,0 - return_price/100
                                  ,refund_time pay_create_time
                                  ,leader_category_code
                                  ,'refund' as type
                              from
                                  ods.ods_of_allot_payment pay_add
                              where
                                  (pay_add.msg_status = 'Y' or pay_add.msg_status = 'y')
                                  and pay_add.order_code is not null
                                  and pay_add.order_code != ''
                          ) t
                      left join (select * from dwd.dwd_order_info WHERE FLAG = 'Y' AND (msg_status = 'Y' or msg_status='y') ) as dd
                          on t.order_code = dd.code
                      )  a
                      left join
                      (
                          select
                              order_code
                              ,min(date(first_pay_time)) as first_pay_date
                              ,sum(if(date(pay_create_time) = date(first_pay_time), payment_price_handle , 0)) as first_day_money
                              ,sum(if(substring(pay_create_time,1,7) = substring(first_pay_time,1,7) , payment_price_handle , 0)) as  first_month_money
                              ,min(first_pay_time) as first_pay_time
                              ,sum(if(pay_create_time = first_pay_time, payment_price_handle , 0)) as  first_pay_money
                              ,max(last_pay_time) as last_pay_time
                              ,sum(payment_price_handle) as sum_pay_money
                          from (
                                  select
                                      order_code
                                      ,pay_create_time
                                      ,payment_price_handle
                                      ,min (pay_create_time) over(partition by order_code) as first_pay_time
                                      ,max (pay_create_time) over(partition by order_code) as last_pay_time
                                  from
                                      dwd.dwd_of_payment
                                  where
                                      FLAG = 'Y'
                                      AND (msg_status = 'Y' or msg_status ='y')
                                      AND order_code <> ''
                                      --AND order_code = '0DP2018052900353'
                              ) pay
                          group by order_code
                      ) b
                      on   a.order_code = b.order_code
                      LEFT JOIN
                          (
                            select
                            category_code,
                            case when num = 1 then '2000-01-01 00:00:00' else  start_time end start_time,
                            end_time,
                            standard_price
                            from(
                            select category_code,start_time,end_time,standard_price,row_number() over(partition by category_code order by start_time asc ) num from dim.dim_become_regular
                              ) as regular
                            ) regular
                      ON a.leader_category_code = regular.category_code
                      where (b.first_pay_time >= regular.start_time
                          AND b.first_pay_time < regular.end_time ) or regular.category_code is null
              ) order_base
              group by  order_code,leader_category_code,return_pay,first_pay_date,first_pay_money,first_day_money,first_month_money,first_pay_time,last_pay_time,sum_pay_money,standard_price

          ) order_pay    on order_pay.order_code = order_info.order_code
        """)

    dws_order_df.write.mode(SaveMode.Overwrite).format("parquet")
        .partitionBy("dt").save(file_path)

    val hivePartitions: util.List[Row] = dws_order_df.select($"dt").distinct().collectAsList()
    hivePartitions.toArray().par.foreach(r => {
      val  Row(dt) = r
      spark.sql(s"ALTER TABLE tmp.dws_order ADD IF NOT EXISTS PARTITION(dt='$dt')")
    })

    spark.stop()
  }
}
